package com.zetyun.tiger.connector.sr.v1;

import com.zetyun.rt.meta.annotation.ActionMeta;
import com.zetyun.rt.sdk.internal.DelegateSinkAction;
import com.zetyun.rt.sdk.operator.OperatorContext;
import lombok.extern.slf4j.Slf4j;
import lombok.val;

/**
 * @author ：zhou gh
 * @date ：2022/01/18 11:32
 */
@ActionMeta(
        id = "dz.flink-StartRocks-http-sink",
        name = "Flink StartRocks Http Sink",
        tags = {"StarRocks", "http", "json"},
        category = "sink",
        parameterClass = RtSrSinkParameter.class,
        description = "参考自 https://github.com/StarRocks/flink-connector-starrocks "

)
@Slf4j
public class RtSrSinkAction extends DelegateSinkAction {

    @Override
    public void init(OperatorContext context) {
        super.init(context);
        RtSrSinkParameter param = context.getParameters();
        log.info("{}  received param: {}", getClass().getSimpleName(), param);

        val srSinkFun = new RtEventStarRocksSinkFunction(param.buildStarRocksSinkOptions());

        try {
            delegate(srSinkFun);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }

    }
}
